package io.github.bigbird0101.datatransfer.container;

import cn.hutool.json.JSONUtil;
import integration.utils.trace.LogUtils;
import io.github.bigbird0101.datatransfer.collector.Communication;
import io.github.bigbird0101.datatransfer.collector.CommunicationTool;
import lombok.extern.slf4j.Slf4j;

import java.text.SimpleDateFormat;

/**
 * 阻塞流处理，读写分别运行于不同线程
 */
@Slf4j
public class StreamTaskContainer extends AbstractContainer {

    private static final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    StreamTaskExecutor taskExecutor;
    private long startTimeStamp;

    private long endTimeStamp;

    private int totalStage = 1;

    private String taskId;

    public StreamTaskContainer(String config) {
        super(config);
        // 任务id
        taskId = JSONUtil.parseObj(config).getStr("taskId");
        taskExecutor = new StreamTaskExecutor(config);
        taskExecutor.init();
    }

    public static StreamTaskContainer getStreamCommonTaskContainer(String config) {
        return new StreamTaskContainer(config);
    }

    @Override
    public void start() {
        this.startTimeStamp = System.currentTimeMillis();
        taskExecutor.start();
    }

    @Override
    public void sendData(String data) {
        taskExecutor.sendData(data);
    }

    @Override
    public void stop() {
        try {

            taskExecutor.stop();
        } finally {
            this.endTimeStamp = System.currentTimeMillis();
            this.logStatistics();
        }
    }

    public void shutdown() {
        taskExecutor.shutdown();
    }


    private Communication reportTaskGroupCommunication(Communication lastTaskGroupContainerCommunication, int taskCount) {
        long currentTimeStamp = System.currentTimeMillis();
        long totalCosts = (currentTimeStamp - this.startTimeStamp) / 1000;
        if (totalCosts == 0L) {
            totalCosts = 1L;
        }

        this.taskExecutor.getTaskCommunication().setTimestamp(System.currentTimeMillis());
        Communication reportCommunication = CommunicationTool.getReportCommunication(this.taskExecutor.getTaskCommunication(),
                lastTaskGroupContainerCommunication, taskCount);

        // 字节速率
        long byteSpeedPerSecond = this.taskExecutor.getTaskCommunication().getLongCounter(CommunicationTool.READ_SUCCEED_BYTES)
                / totalCosts;

        long recordSpeedPerSecond = this.taskExecutor.getTaskCommunication().getLongCounter(CommunicationTool.READ_SUCCEED_RECORDS)
                / totalCosts;

        lastTaskGroupContainerCommunication.setLongCounter(CommunicationTool.BYTE_SPEED, byteSpeedPerSecond);
        lastTaskGroupContainerCommunication.setLongCounter(CommunicationTool.RECORD_SPEED, recordSpeedPerSecond);

        LogUtils.info(log, "taskId: " + taskId + ", " + CommunicationTool.Stringify.getSnapshot(this.taskExecutor.getTaskCommunication()));
        return reportCommunication;
    }


    private void logStatistics() {
        long totalCosts = (this.endTimeStamp - this.startTimeStamp) / 1000;
        if (totalCosts == 0L) {
            totalCosts = 1L;
        }
        Communication communication = this.taskExecutor.getTaskCommunication();
        communication.setTimestamp(this.endTimeStamp);

        Communication tempComm = new Communication();
        tempComm.setTimestamp(this.endTimeStamp);

        Communication reportCommunication = CommunicationTool.getReportCommunication(communication, tempComm, this.totalStage);

        // 字节速率
        long byteSpeedPerSecond = communication.getLongCounter(CommunicationTool.READ_SUCCEED_BYTES)
                / totalCosts;

        long recordSpeedPerSecond = communication.getLongCounter(CommunicationTool.READ_SUCCEED_RECORDS)
                / totalCosts;

        reportCommunication.setLongCounter(CommunicationTool.BYTE_SPEED, byteSpeedPerSecond);
        reportCommunication.setLongCounter(CommunicationTool.RECORD_SPEED, recordSpeedPerSecond);


        LogUtils.info(log, String.format(
                "\n任务ID:" + taskId +
                        "\n" + "%-26s: %-18s\n" + "%-26s: %-18s\n" + "%-26s: %19s\n"
                        + "%-26s: %19s\n" + "%-26s: %19s\n" + "%-26s: %19s\n"
                        + "%-26s: %19s\n",
                "任务启动时刻",
                dateFormat.format(startTimeStamp),

                "任务结束时刻",
                dateFormat.format(endTimeStamp),

                "任务总计耗时",
                String.valueOf(totalCosts) + "s",
                "任务平均流量",
                CommunicationTool.Stringify.stringify(byteSpeedPerSecond)
                        + "/s",
                "记录写入速度",
                String.valueOf(recordSpeedPerSecond)
                        + "rec/s", "读出记录总数",
                String.valueOf(CommunicationTool.getTotalReadRecords(communication)),
                "读写失败总数",
                String.valueOf(CommunicationTool.getTotalErrorRecords(communication))
        ));

        if (communication.getLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS) > 0
                || communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS) > 0
                || communication.getLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS) > 0) {
            LogUtils.info(log, String.format(
                    "\n" + "%-26s: %19s\n" + "%-26s: %19s\n" + "%-26s: %19s\n",
                    "Transformer成功记录总数",
                    communication.getLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS),

                    "Transformer失败记录总数",
                    communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS),

                    "Transformer过滤记录总数",
                    communication.getLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS)
            ));
        }


    }
}
